package com.fulihui.information.core.zbus;

import com.alibaba.fastjson.JSON;
import com.fulihui.information.biz.content.SpiderDataProcessBiz;
import com.fulihui.information.common.config.AppConst;
import com.fulihui.information.request.SpiderDataUploadRequest;
import org.apache.commons.lang3.StringUtils;
import org.near.toolkit.common.Log;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.zbus.broker.Broker;
import org.zbus.broker.ZbusBroker;
import org.zbus.mq.Consumer;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;

/**
 * @author leeson 2016年5月6日 下午7:21:43
 *         短消息队列 消费者 （随容器启动）
 */
@Component
public class ZbusConsumerHandle {
    @Autowired
    private AppConst             appConst;
    @Autowired
    private SpiderDataProcessBiz spiderDataProcessBiz;
    private Broker               broker;

    public static final int PROCESSOR_COUNT = 2 * Runtime.getRuntime().availableProcessors();

    @PostConstruct
    public void init() throws Exception {
        broker = new ZbusBroker(appConst.getZbusUrl());

        for (int i = 0; i < PROCESSOR_COUNT; i++) {
            new Consumer(broker, Const.QUEUE_NAME).start((msg, consumer) -> {
                try {
                    String messageJson = msg.getBodyString();
                    Log.info("S<-Q: " + JSON.parseObject(messageJson).getString("trackNo"));

                    if (StringUtils.isNotEmpty(messageJson)) {
                        spiderDataProcessBiz
                                .process(JSON.parseObject(messageJson, SpiderDataUploadRequest.class));
                    }
                } catch (Exception e) {
                    Log.error(e.getMessage(), e);
                }
            });
        }
    }

    @PreDestroy
    public void destroy() {
        try {
            broker.close();
        } catch (IOException e) {
            Log.error(e.getMessage(), e);
        }
    }
}